package com.atguigu.tingshu.user.service.impl;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.MongoUtil;
import com.atguigu.tingshu.model.user.UserListenProcess;
import com.atguigu.tingshu.user.service.UserListenProcessService;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.atguigu.tingshu.vo.user.UserListenProcessVo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Service
@SuppressWarnings({"all"})
public class UserListenProcessServiceImpl implements UserListenProcessService {

	@Autowired
	private MongoTemplate mongoTemplate;

    @Autowired
    KafkaService kafkaService;

    @Override
    public BigDecimal getTrackBreakSecond(Long userId, Long trackId) {
        // 根据userId和trackId查询用户的播放进度
        //构建查询条件
        Query query = new Query();
        query.addCriteria(Criteria.where("userId").is(userId).and("trackId").is(trackId));
        //最新的一条
        query.with(Sort.by(Sort.Direction.DESC, "updateTime"));
        query.limit(1);
        UserListenProcess userListenProcess = mongoTemplate.findOne(query, UserListenProcess.class, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId));
        if (userListenProcess != null) {
            //返回播放进度
            return userListenProcess.getBreakSecond();
        }

        return new BigDecimal("0.00");
    }

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 更新声音进度
     *
     * @param userId              用户id
     * @param userListenProcessVo 用户监听进程vo
     */
    @Override
    public void updateListenProcess(Long userId, UserListenProcessVo userListenProcessVo) {
        if (userListenProcessVo != null) {
            //根据userId和trackId查询用户的播放进度
            //构建查询条件
            Query query = new Query();
            query.addCriteria(Criteria.where("userId").is(userId).and("trackId").is(userListenProcessVo.getTrackId()));
            query.limit(1);
            UserListenProcess userListenProcess = mongoTemplate.findOne(query, UserListenProcess.class, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId));
            if (userListenProcess == null) {
                //如果用户的播放进度不存在，就新增一条
                userListenProcess = new UserListenProcess();
                userListenProcess.setAlbumId(userListenProcessVo.getAlbumId());
                userListenProcess.setUserId(userId);
                userListenProcess.setTrackId(userListenProcessVo.getTrackId());
                userListenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
                userListenProcess.setIsShow(1);
                userListenProcess.setCreateTime(DateTime.now());
                userListenProcess.setUpdateTime(DateTime.now());
                mongoTemplate.save(userListenProcess, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId));
            } else {
                //如果用户的播放进度存在，就更新一条
                userListenProcess.setBreakSecond(userListenProcessVo.getBreakSecond());
                userListenProcess.setAlbumId(userListenProcessVo.getAlbumId());
                userListenProcess.setUpdateTime(DateTime.now());
                mongoTemplate.save(userListenProcess, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId));
            }
            String key = RedisConstant.USER_TRACK_REPEAT_STAT_PREFIX + userId + ":" + userListenProcessVo.getTrackId();
            long time = DateUtil.endOfDay(new Date()).getTime() - System.currentTimeMillis();

            //如果用户的播放进度存在，就更新一条
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(key, userListenProcess.getTrackId(), time, TimeUnit.MILLISECONDS);
            if (flag) {
                //5.如果是首次更新播放进度，发送消息到Kafka话题
                //5.1 构建更新声音播放进度MQVO对象
                TrackStatMqVo trackStatMqVo = new TrackStatMqVo();
                trackStatMqVo.setBusinessNo(IdUtil.fastSimpleUUID());
                trackStatMqVo.setAlbumId(userListenProcess.getAlbumId());
                trackStatMqVo.setTrackId(userListenProcess.getTrackId());
                trackStatMqVo.setStatType(SystemConstant.TRACK_STAT_PLAY);
                trackStatMqVo.setCount(1);
                kafkaService.sendMessage(KafkaConstant.QUEUE_TRACK_STAT_UPDATE, JSON.toJSONString(trackStatMqVo));

                //生成唯一业务标识

            }
        }
    }

    /**
     * 获取最近的播放声音
     *
     * @param userId 用户id
     * @return 列表<map < string ， 对象>>
     */
    @Override
    public Map<String, Object> getLatelyTrack(Long userId) {
        Query query = new Query();
        Query query1 = query.addCriteria(Criteria.where("userId").is(userId));
        query1.with(Sort.by(Sort.Direction.DESC, "updateTime"));
        query1.limit(1);
        HashMap<String, Object> stringObjectHashMap = new HashMap<>();
        UserListenProcess one = mongoTemplate.findOne(query1, UserListenProcess.class, MongoUtil.getCollectionName(MongoUtil.MongoCollectionEnum.USER_LISTEN_PROCESS, userId));
        if (one != null) {

            stringObjectHashMap.put("trackId", one.getTrackId());
            stringObjectHashMap.put("albumId", one.getAlbumId());
        }
        return stringObjectHashMap;
    }

}
